package com.imooc.hadoop.mapreduce.topN;

import com.imooc.hadoop.mapreduce.wordsCount.WordsCountReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @program: hadoop
 * @description: TopNReducer
 * @author: zengchen
 * @create: 2021-03-30 15:20
 **/
public class TopNReducer extends Reducer<LongWritable, NullWritable, LongWritable, NullWritable> {

    Logger logger = LoggerFactory.getLogger(TopNReducer.class);

    private List keys;


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        if (keys == null) {
            keys = new ArrayList<>();
        }
    }

    @Override
    protected void reduce(LongWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        logger.info("reduce key={}", key.get());
        keys.add(key.get());
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        Configuration configuration = context.getConfiguration();
        String topN = configuration.get(TopNJob.KEY_NAME);
        logger.info("topN value={}", topN);
        int nums = 1;
        for (int i = keys.size() - 1; i >= 0; i--) {
            if (nums > Integer.parseInt(topN)) {
                break;
            }
            context.write(new LongWritable((Long) keys.get(i)), NullWritable.get());
            nums++;
        }
    }
}
